package com.htdata.iiot.api.hbase.utils;

import com.htdata.iiot.api.tsdb.config.Configs;
import com.htiiot.common.config.ConfigFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author chensi
 */
public class HbaseOperation {
    private static Logger LOGGER = Logger.getLogger(HbaseOperation.class);

    public static Configuration conf = null;

    static
    {
        conf = HBaseConfiguration.create();
        Properties pro = ConfigFactory.getConfigMgr().config(Configs.TSDBConfig.CLIENT_CONFIG_SECTION).getAll();
        conf.set("hbase.zookeeper.quorum",pro.getProperty("hbase.zookeeper.quorum"));
        conf.set("zookeeper.znode.parent", pro.getProperty("hbase.zookeeper.znode.parent"));
    }

    /**
     * 创建表
     */
    public static void creatTable(String tableName, String[] familys)
            throws Exception
    {

        HBaseAdmin admin = new HBaseAdmin(conf);
        if (admin.tableExists(tableName))
        {
            System.out.println("table already exists!");
        }
        else
        {
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);
            for (int i = 0; i < familys.length; i++)
            {
                tableDesc.addFamily(new HColumnDescriptor(familys[i]));
            }
            admin.createTable(tableDesc);
            System.out.println("create table " + tableName + " ok.");
        }
    }

    /**
     * 删除表
     */
    public static void deleteTable(String tableName)
            throws Exception
    {

        try
        {
            HBaseAdmin admin = new HBaseAdmin(conf);
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("delete table " + tableName + " ok.");
        }
        catch (MasterNotRunningException e)
        {
            e.printStackTrace();
        }
        catch (ZooKeeperConnectionException e)
        {
            e.printStackTrace();
        }
    }

    /**
     * 添加记录
     */
    public static void addRecord(String tableName, String rowKey, String family, String qualifier,
                                 String value)
            throws Exception
    {

        try
        {
            HTable table = new HTable(conf, tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
            System.out.println("insert recored " + rowKey + " to table " + tableName + " ok.");
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
    }

    /**
     * 删除记录
     */
    public static void delRecord(String tableName, String rowKey)
            throws IOException
    {

        HTable table = new HTable(conf, tableName);
        List list = new ArrayList();
        Delete del = new Delete(rowKey.getBytes());
        list.add(del);
        table.delete(list);
        System.out.println("del recored " + rowKey + " ok.");
    }

    public static Result getResult(String tableName,String rowKey) throws IOException {
        HTable table = new HTable(conf, tableName);
        Get get = new Get(rowKey.getBytes());
        Result rs = table.get(get);
        return rs;
    }

    /**
     * 一次性查询多个rowkey返回结果数组
     * @param tableName
     * @param rowKeys
     * @return
     * @throws IOException
     */
    public static Result[] getResults(String tableName,String[] rowKeys) throws IOException {
        HTable table = new HTable(conf,tableName);
        List<Get> getL = new ArrayList<>();
        for(String rowKey:rowKeys){
            Get get = new Get(rowKey.getBytes());
            getL.add(get);
        }
        Result[] results = table.get(getL);
        return results;
    }

    /**
     * 获取某条记录
     */
    public static void getOneRecord(String tableName, String rowKey)
            throws IOException
    {

        HTable table = new HTable(conf, tableName);
        Get get = new Get(rowKey.getBytes());
        Result rs = table.get(get);
        for(Cell cell:rs.rawCells()){
            System.out.println(
                    "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
                            "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
                            "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
                            "值:"   + new String(CellUtil.cloneValue(cell)) + "\t" +
                            "时间戳:" + cell.getTimestamp());
        }
    }

    /**
     * 获取某条记录
     */
    public static void getOneRecord(String tableName, byte[] rowKey)
            throws IOException
    {

        HTable table = new HTable(conf, tableName);
        Get get = new Get(rowKey);
        Result rs = table.get(get);
        int count = 0;
        for(Cell cell:rs.rawCells()){
            count++;
            /*System.out.println(
                    "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
                            "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
                            "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
                            "值:"   + new String(CellUtil.cloneValue(cell)) + "\t" +
                            "时间戳:" + cell.getTimestamp());*/
        }
        System.out.println("all:"+count);
    }

    /**
     * 通过metric查询row_key
     * @param tableName
     * @param metric
     * @return
     * @throws IOException
     */
    public static byte[] getRowKeyFromMetric(String tableName, byte[] metric)
            throws IOException {

        HTable table = new HTable(conf, tableName);
        Get get = new Get(metric);
        Result rs = table.get(get);
        for (Cell cell : rs.rawCells()) {
            if ("id".equals(new String(CellUtil.cloneFamily(cell))) && "metrics".equals(new String(CellUtil.cloneQualifier(cell)))) {
                return CellUtil.cloneValue(cell);
            }
        }
        return null;
    }

    /**
     * 获取所有记录
     */
    public static void getAllRecord(String tableName)
    {

        try
        {
            HTable table = new HTable(conf, tableName);
            Scan s = new Scan();
            ResultScanner ss = table.getScanner(s);
            for (Result r : ss)
            {
                for (KeyValue kv : r.raw())
                {
                    System.out.print(new String(kv.getRow()) + " ");
                    System.out.print(new String(kv.getFamily()) + ":");
                    System.out.print(new String(kv.getQualifier()) + " ");
                    System.out.print(kv.getTimestamp() + " ");
                    System.out.println(new String(kv.getValue()));
                }
            }
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
    }

    //批量删除Hbase中的数据
    public static void batchDel(String tableName,List<byte[]> list) throws IOException {
        HTable table = new HTable(conf,tableName);
        List<Delete> deleteL = new ArrayList<>();
        for(byte[] b:list){
            Delete d = new Delete(b);
            deleteL.add(d);
        }
        table.delete(deleteL);
    }

}
